package cn.com.bonc.khjy.schedu;

import cn.com.bonc.khjy.utils.FileUtil;
import cn.com.bonc.khjy.utils.FtpUtil;
import cn.com.bonc.khjy.utils.SystemMessageId;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author BTW
 */
@Component
public class FileMonitor {

    private final static Log log = LogFactory.get();
    private static FtpUtil ftpUtil = new FtpUtil();
    @Value( "${temp.file.store.path}" )
    private String tempFileDirectory;
    @Value( "${max.file.line}" )
    private int maxLine;
    @Value( "${file.store.path}" )
    private String fileDirectory;
    @Value( "${ftp.remote.dir}" )
    private String ftpRemoteDir;
    @Value( "${ftp.server.ip}" )
    private String ftpServerIp;
    @Value( "${ftp.server.port}" )
    private int ftpServerPort;
    @Value( "${ftp.username}" )
    private String ftpUsername;
    @Value( "${ftp.password}" )
    private String ftpPassword;
    @Value( "${ftp.local.dir}" )
    private String ftpLocalDir;
    private volatile boolean tempFileStop = false;
    private volatile boolean ftpFileStop = false;

    private volatile LinkedBlockingQueue< String > queue = new LinkedBlockingQueue<>( 1000 );
    private Map< String, String > tempFileMap = new ConcurrentHashMap<>();
    private volatile LinkedBlockingQueue< Map > ftpFileQueue = new LinkedBlockingQueue<>( 20 );

    private String getNewTempFilePath() {
        String date = new SimpleDateFormat( "yyyyMMddHHmmss" ).format( new Date() );
        return Paths.get( tempFileDirectory, date + ".temp" ).toString();
    }

    private boolean uploadFtpFile( String remotePath, String filePath, String targetFileName ) {
        try {
            log.trace( "连接FTP服务器..." );
            boolean connected = ftpUtil.connect( ftpServerIp, ftpServerPort, ftpUsername, ftpPassword );
            int retryTimes = 0;
            boolean isSuccess = false;
            log.trace( "连接FTP服务器...{}", connected ? "成功" : "失败" );
            if ( connected ) {
                isSuccess = ftpUtil.uploadFile( remotePath, filePath, targetFileName );
                while ( !isSuccess ) {
                    log.debug("上传文件：{}到ftp失败，重试次数：{}", targetFileName, retryTimes );
                    isSuccess = ftpUtil.uploadFile( remotePath, filePath, targetFileName );
                    retryTimes++;
                    if ( retryTimes >= 3 ) {
                        log.debug( "上传文件：{}到ftp失败，重试次数：{}，放弃上传", targetFileName, retryTimes );
                        break;
                    }
                }
            }
            return isSuccess;
        } catch ( Exception e ) {
            log.error( e.getMessage(), e );
            return false;
        }
    }

    public void start() {
        ThreadUtil.execAsync( () -> {
            while ( !tempFileStop ) {
                try {
                    log.trace( "进入临时文件写入线程" );
                    String input = queue.take();
                    log.trace( "读取到数据：{}", input );
                    String filePath = tempFileMap.get( "filePath" );
                    if ( StrUtil.isBlank( filePath ) ) {
                        filePath = getNewTempFilePath();
                        tempFileMap.put( "filePath", filePath );
                        log.trace( "新文件名为：{}", filePath );
                        FileUtil.createFile( filePath );
                    }
                    File file = new File( filePath );
                    FileUtil.saveFile( input, file );
                    int count = FileUtil.getFileLineCount( file );
                    if ( count >= maxLine ) {
                        log.trace( "文件行数：{}", count  );
                        TimeUnit.SECONDS.sleep( 1 );
                        if ( FileUtil.moveFile( fileDirectory, filePath ) ) {
                            String ftpFileName = filePath.substring( filePath.lastIndexOf( File.separator ), filePath.length() ).replace( ".temp", ".txt" );
                            String ftpFilePath = Paths.get( fileDirectory, ftpFileName ).toString();
                            String qcFilePath = ftpFilePath.replace( ".txt", ".qc" );
                            Map< String, String > fileMap = new ConcurrentHashMap<>();
                            fileMap.put( "ftpFile", ftpFilePath );
                            fileMap.put( "qcFile", qcFilePath );
                            ftpFileQueue.put( fileMap );
                            log.trace( "重置临时文件" );
                            tempFileMap.put( "filePath", "" );
                        } else {
                            log.trace( "移动文件失败"  );
                        }
                    }
                } catch ( InterruptedException e ) {
                    log.error( e, SystemMessageId.CMP0705, Thread.currentThread().getName() );
                }
            }
        } );

        ThreadUtil.execAsync( () -> {
            while ( !ftpFileStop ) {
                try {
                    log.trace( "进入ftp文件上传线程" );
                    Map< String, String > fileMap = ftpFileQueue.take();
                    log.trace( "获取到待上传文件：{}", fileMap );
                    String ftpFilePath = fileMap.get( "ftpFile" );
                    String qcFilePath = fileMap.get( "qcFile" );
                    if ( StrUtil.isNotBlank( ftpFilePath ) && StrUtil.isNotBlank( qcFilePath ) ) {
                        int count = FileUtil.getFileLineCount( new File( ftpFilePath ) );
                        String input = ftpFilePath + " : " + count;
                        FileUtil.createFile( qcFilePath );
                        File qcFile = new File( qcFilePath );
                        File ftpFile = new File( ftpFilePath );
                        FileUtil.saveFile( input, qcFile );

                        boolean ftpSuccess = uploadFtpFile( ftpRemoteDir, ftpFilePath, ftpFile.getName() );
                        boolean qcSuccess = uploadFtpFile( ftpRemoteDir, qcFilePath, qcFile.getName() );

                        if ( !ftpSuccess && qcSuccess ) {
                            ftpFileQueue.put( fileMap );
                        }
                    }
                } catch ( InterruptedException e ) {
                    log.error( e, SystemMessageId.CMP0705, Thread.currentThread().getName() );
                }
            }
        } );
    }

    public void toStop() {
        tempFileStop = true;
        ftpFileStop = true;
    }

    public void addKafkaInfo( String input ) {
        try {
            log.trace( "获取到数据:{}", input );
            if ( StrUtil.isNotBlank( input ) ) {
                queue.put( input );
                log.trace( "消息队列：{}", queue );
            }
        } catch ( InterruptedException e ) {
            log.error( e, SystemMessageId.CMP0705, Thread.currentThread().getName() );
        }
    }

}
